<?php
/*
 * @Create By: Vscode
 * @Author: falcon
 * @Date: 2020-01-01 16:47:12
 * kafka 生产者 和 消费者
 */
namespace application\controller;

use application\depend\Applog;
use application\depend\Dbkfk;

class Kafka{

    public function main(){
        // $this->producer();
        $this->consumer();
    }// main func

    // 生产者
    public function producer(){
        $broker_list = '127.0.0.1:9092';
        $topic = 'test';
        $partition = 0;

        $rk = new \RdKafka\Producer();

        $rk->setLogLevel(LOG_DEBUG);

        $rk->addBrokers($broker_list);

        $topic = $rk->newTopic($topic);
        $message = 'this is a php test' . date('Y-m-d H:i:s');
        $topic->produce(RD_KAFKA_PARTITION_UA,$partition,json_encode($message));
    }// 

    // 消费者
    public function consumer(){
        $conf = new \RdKafka\Conf();

        $conf->set('group.id', 'myConsumerGroup');

        $rk = new \RdKafka\Consumer($conf);
        $rk->addBrokers("localhost:9092");

        $topicConf = new \RdKafka\TopicConf();
        $topicConf->set('auto.commit.interval.ms', 100);
        $topicConf->set('offset.store.method', 'file');
        $topicConf->set('offset.store.path', sys_get_temp_dir());
        $topicConf->set('auto.offset.reset', 'smallest');

        $topic = $rk->newTopic("test", $topicConf);

        // Start consuming partition 0
        $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

        while (true) {
            $message = $topic->consume(0, 120*10000); // 分区，超时时间
            switch ($message->err) {
                case RD_KAFKA_RESP_ERR_NO_ERROR:
                //没有错误打印信息
                    var_dump($message);

                    if(!empty($message) && is_object($message)){
                        // $message = array($message); // 强转数组
                        // $message = array($message[0]); // 强转数组
                        $msg = $message->payload;
                        var_dump($msg);
                        //写入数据库  只接收topic = test 的数据
                        if(!empty($msg)){
                            $pdo = Dbkfk::getInstance();
                            $sql = "INSERT INTO kafka_info (message) values ('$msg')";
                            $pre = $pdo->prepareSql($sql);

                            try{
                                if(!$pre->execute()){
                                    $e = [
                                        'title' => 'Kafka录入信息失败',
                                        'sql' => $sql,
                                        'pdo' => $pdo->presqlErrInfo(),
                                    ];
                                    Applog::errorLog($e);
                                    var_dump('数据库写入失败');
                                    return;
                                }
                                var_dump('已写入数据库');


                            }catch(\Exception $e){
                                $e = [
                                    'title' => 'Kafka录入信息失败 e-catch',
                                    'sql' => $sql,
                                    'pdo' => $pdo->presqlErrInfo(),
                                ];
                                Applog::errorLog($e);
                            }
                        }


                    }
                    break;
                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                    echo "等待接收信息\n";
                    break;
                case RD_KAFKA_RESP_ERR__TIMED_OUT:
                    echo "超时\n";
                    break;
                default:
                    throw new \Exception($message->errstr(), $message->err);
                    break;
            }// switch
        }// while
    }// func
}// func

